# utils/task_executor.py
import asyncio
from concurrent.futures import ThreadPoolExecutor
from django.utils import timezone

from app.models import TaskExecutionLog
import logging
logger = logging.getLogger(__name__)

class AsyncTaskExecutor:
    """异步任务执行器"""
    _instance = None
    _executor = None  # 初始化为 None
    _tasks = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._tasks = {}
            # 延迟初始化线程池
            cls._init_executor()
        return cls._instance

    @classmethod
    def _init_executor(cls):
        """初始化线程池"""
        if cls._executor is None or cls._executor._shutdown:
            # 创建新的线程池
            max_workers = 10  # 根据需求调整
            cls._executor = ThreadPoolExecutor(max_workers=max_workers)
            print(f"初始化线程池，最大工作线程数: {max_workers}")
    def submit_task(self, strategy_id, task_func, *args):
        """提交任务到线程池"""
        try:
            # 创建任务日志
            log = TaskExecutionLog.objects.create(
                strategy_id=strategy_id,
                status=TaskExecutionLog.StatusChoices.PENDING
            )
            print(f"创建任务日志: 日志ID={log.id}, 策略ID={strategy_id}")

            if asyncio.iscoroutinefunction(task_func):
                # 异步任务需要特殊处理
                future = self._executor.submit(
                    self._run_async_task_wrapper,
                    task_func,
                    log.id,
                    *args
                )
            else:
                # 同步任务
                future = self._executor.submit(
                    self._run_task,
                    task_func,
                    log.id,
                    *args
                )

            # 存储任务信息
            self._tasks[log.id] = {
                'future': future,
                'log_id': log.id,
                'strategy_id': strategy_id,
                'start_time': timezone.now(),
                'status': 'pending'
            }
            print(f"已存储任务信息: 日志ID={log.id}, 策略ID={strategy_id}")

            return log.id
        except Exception as e:
            print(f"提交任务失败: {e}")
            return None

    def _run_task(self, task_func, log_id, *args):
        """实际执行任务的方法"""
        try:
            print(f"🔧 开始执行任务: 日志ID={log_id}")

            # 更新任务状态为运行中
            self._update_task_status(log_id, 'running')
            print(f"🔄 任务状态更新为 running: 日志ID={log_id}")

            try:
                # 执行任务
                print(f"⚡ 调用任务函数: 日志ID={log_id}")
                result = task_func(*args)
                print(f"✅ 任务执行成功: 日志ID={log_id}, 结果={result}")

                # 更新任务状态为已完成
                self._update_task_status(log_id, 'completed', result)
                print(f"🔄 任务状态更新为 completed: 日志ID={log_id}")
                return result
            except Exception as e:
                print(f"❌ 任务执行失败: 日志ID={log_id}, 错误={e}")
                # 更新任务状态为失败
                self._update_task_status(log_id, 'failed', error=str(e))
                print(f"🔄 任务状态更新为 failed: 日志ID={log_id}")
                raise
        except Exception as e:
            print(f"❌ _run_task 内部错误: {e}")

    async def _run_async_task(self, task_func, log_id, *args):
        """执行异步任务"""
        try:
            self._update_task_status(log_id, 'running')

            # 执行异步任务
            result = await task_func(*args)
            print(f"异步任务result:{result}")
            # 确保结果可序列化
            if hasattr(result, '__await__'):
                result = await result
            serializable_result = self._make_json_serializable(result)
            print(f"异步任务执行完成:更新completed,日志ID:{log_id}")
            self._update_task_status(log_id, 'completed', result=serializable_result)
            return result
        except Exception as e:
            self._update_task_status(log_id, 'failed', error=str(e))
            raise

    def _run_async_task_wrapper(self, *args):
        """在新事件循环中运行异步任务"""
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            return loop.run_until_complete(self._run_async_task(*args))
        finally:
            loop.close()

    def _make_json_serializable(self, data):
        """递归确保数据可JSON序列化"""
        if data is None:
            return None
        if isinstance(data, (str, int, float, bool)):
            return data
        if isinstance(data, dict):
            return {k: self._make_json_serializable(v) for k, v in data.items()}
        if isinstance(data, (list, tuple)):
            return [self._make_json_serializable(item) for item in data]
        return str(data)  # 其他类型转为字符串

    def _update_task_status(self, log_id, status, result=None, error=None):
        """安全更新状态"""
        try:
            # 处理结果数据
            safe_result = None
            if result is not None:
                if asyncio.iscoroutine(result):
                    print("尝试序列化协程对象，应先await！")
                    safe_result = "<coroutine>"
                else:
                    safe_result = self._make_json_serializable(result)

            # 更新内存状态
            if log_id in self._tasks:
                self._tasks[log_id].update({
                    'status': status,
                    'end_time': timezone.now(),
                    'result': safe_result,
                    'error': str(error)[:2000] if error else None
                })

            # 更新数据库
            log = TaskExecutionLog.objects.filter(id=log_id).first()
            if log:
                log.status = status
                log.result = {'result': safe_result} if safe_result else None
                log.error_message = str(error)[:2000] if error else None
                if status in ['completed', 'failed']:
                    log.end_time = timezone.now()
                log.save()
        except Exception as e:
            print(f"状态更新失败 {log_id}: {e}", exc_info=True)
    def get_task_status(self, task_id):
        """获取任务状态"""
        return self._tasks.get(task_id, {'status': 'unknown'})

    def get_all_tasks(self):
        """获取所有任务状态"""
        return self._tasks.values()